package com.xy.job.executor.core

import com.xy.job.executor.config.AdminProperties
import com.xy.job.executor.config.ExecutorProperties
import com.xy.job.executor.core.client.DefaultAdminClient
import com.xy.job.executor.core.client.IAdminClient
import com.xy.job.executor.core.job.ExecutorCoroutineScope
import com.xy.job.executor.core.job.IJobHandler
import com.xy.job.executor.core.job.JobHandlerRegistry
import com.xy.job.executor.core.server.EmbedServer
import com.xy.job.executor.core.server.ExecutorServiceBiz
import com.xy.job.executor.core.server.RegisterThread
import org.slf4j.LoggerFactory
import java.io.InputStreamReader
import java.nio.charset.StandardCharsets
import java.util.*

class Executor(
    private val configurationFilePath: String
) {
    private lateinit var adminClient: IAdminClient
    private lateinit var executorServiceBiz: ExecutorServiceBiz
    private lateinit var embedServer: EmbedServer
    private lateinit var registerThread: RegisterThread

    private val log = LoggerFactory.getLogger(this.javaClass)

    private val shutdownHook = {
        log.info("ready to shutdown executor")
        // 停止注册线程
        registerThread.cancel()
        // 发送下线通知
        adminClient.removeRegister()
        // 停止协程作业
        ExecutorCoroutineScope.cancelJobs()
        log.info("executor shutdown successfully")
    }

    fun start(wait: Boolean = true) {
        val properties = loadPropertiesFile(configurationFilePath)
        val adminProperties = loadAdminProperties(properties)
        val executorProperties = loadExecutorProperties(properties)
        adminClient = DefaultAdminClient(adminProperties.host, adminProperties.port.toInt())
        executorServiceBiz = ExecutorServiceBiz(adminClient)
        embedServer = EmbedServer(executorProperties.port.toInt(), executorServiceBiz)
        registerThread = RegisterThread(adminClient = adminClient, executorProperties = executorProperties)

        registerThread.start()
        embedServer.addShutdownHook(shutdownHook)
        embedServer.start(wait)
    }

    fun register(vararg jobHandlers: IJobHandler) {
        jobHandlers.forEach {
            JobHandlerRegistry.register(it::class.qualifiedName!!, it)
        }
    }

    private fun loadAdminProperties(properties: Properties): AdminProperties {
        return AdminProperties().apply {
            host = properties.getProperty("xy-job.admin.host")
            port = properties.getProperty("xy-job.admin.port")
        }
    }

    private fun loadExecutorProperties(properties: Properties): ExecutorProperties {
        return ExecutorProperties().apply {
            ip = properties.getProperty("xy-job.executor.ip")
            port = properties.getProperty("xy-job.executor.port")
            groupCode = properties.getProperty("xy-job.executor.executor-name")
            logDir = properties.getProperty("xy-job.executor.log-path")
            accessToken = properties.getProperty("xy-job.executor.access-token")
        }
    }

    private fun loadPropertiesFile(filepath: String): Properties {
        val classLoader = this.javaClass.classLoader
        val propertiesFileInputStream = classLoader.getResourceAsStream(filepath)
            ?: throw RuntimeException("load $filepath error")
        val properties = propertiesFileInputStream.use {
            val properties = Properties()
            val reader = InputStreamReader(propertiesFileInputStream, StandardCharsets.UTF_8)
            properties.load(reader)
            return@use properties
        }
        return properties
    }
}